package com.webull.information.center.carwler.job;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.webull.information.center.carwler.common.model.CollectRecord;
import com.webull.information.center.carwler.common.model.parm.CollectParm;
import com.webull.information.center.carwler.common.util.eljob.ElasticJobUtil;
import com.webull.information.center.carwler.dao.CollectRecordDao;
import com.webull.information.center.carwler.task.News_Bloomberg_Stocks_Task;
import com.webull.securities.client.model.Exchange;
import com.webull.securities.client.model.parm.ExchangeParam;
import com.webull.securities.client.service.ExchangeService;

/**
 * 彭博聚合新闻采集
 * 
 * @author smj
 *
 */
@Component
public class News_Bloomberg_Stock_Job implements SimpleJob {
	protected final Logger logger = LogManager.getLogger(getClass());
	private @Autowired News_Bloomberg_Stocks_Task news_Bloomberg_Stocks_Task;
	private @Autowired ExchangeService exchangeService;
	private @Autowired CollectRecordDao collectRecordDao;
	private final String collectJob = "News_Bloomberg_Stock_Job";

	// US/ CH/ HK/ SP/ JP/ IN
	@SuppressWarnings({ "rawtypes", "serial", "unchecked" })
	private List<BloomTag> processRegions = new ArrayList() {
		{
			add(BloomTag.produce(1, "CH"));
			add(BloomTag.produce(2, "HK"));
			add(BloomTag.produce(5, "JP"));// 日本
			add(BloomTag.produce(6, "US"));
			add(BloomTag.produce(12, "IN"));// 印度
			add(BloomTag.produce(13, "SP"));// 新加坡
		}
	};

	@Override
	public void execute(ShardingContext context) {
		int sharding = Optional.ofNullable(context).map(v -> v.getShardingItem()).orElse(-1);
		int totalCount = Optional.ofNullable(context).map(v -> v.getShardingTotalCount()).orElse(-1);

		logger.info("News_Bloomberg_Stock_Job start:" + ArrayUtils.toString(sharding) + ",totalCount:" + totalCount);

		List<BloomTag> shardBloomTag = Optional.ofNullable(ElasticJobUtil.shardingItems(context, processRegions))
				.orElse(new ArrayList<>(processRegions));
		Collections.shuffle(shardBloomTag);

		for (BloomTag tgs : shardBloomTag) {

			ExchangeParam exchangeParam = new ExchangeParam();
			exchangeParam.setRegionId(tgs.getWbRegionId());
			List<Exchange> exchanges = exchangeService.search(exchangeParam);
			logger.info("News_Bloomberg_Stock_Job div size{};exchanges:{};", exchanges != null ? exchanges.size() : 0,
					ArrayUtils.toString(exchanges));

			process_0(exchanges, tgs.getBbRegionAlias());
		}
	}


	public void process_0(List<Exchange> exchanges, String bbRegionAlias) {
		Collections.shuffle(exchanges);
		logger.info("News_Bloomberg_Stock_Job div size{};shardExchanges:{};", exchanges != null ? exchanges.size() : 0,
				ArrayUtils.toString(exchanges));
		for (int i = 0; exchanges != null && i < exchanges.size(); i++) {
			try {
				List<CollectRecord> records = collectRecordDao.search(collectParm(exchanges.get(i)));
				Integer lastId = getLastTickeId(records);
				logger.info("News_Bloomberg_Stock_Job shardExchanges start collectJob{} div ,exchange:{},lastId{}",
						collectJob, exchanges.get(i).getExchangeCode(), lastId);
				Integer lastId_ = news_Bloomberg_Stocks_Task
						.stockFinanceNewsExchangeCode(exchanges.get(i).getExchangeCode(), bbRegionAlias);
				CollectRecord record = backRecord(exchanges.get(i));
				record.setLastTickerId(lastId_);
				Integer collId = collectRecordDao.add(record);
				logger.info("News_Bloomberg_Stock_Job shardExchanges end collectJob{} div ,exchange:{},collId{}",
						collectJob, exchanges.get(i).getExchangeCode(), collId);
			} catch (Exception e) {
				logger.error("News_Bloomberg_Stock_Job shardExchanges collectJob div error:" + exchanges.get(i), e);
			}
		}
	}

	private Integer getLastTickeId(List<CollectRecord> records) {
		if (records == null || records.isEmpty()) {
			return -1;
		} else {
			return records.get(0).getLastTickerId();
		}
	}

	private CollectParm collectParm(Exchange exchange) {
		CollectParm parm = new CollectParm();
		parm.setCollectJob(collectJob);
		parm.setExchangeCode(exchange.getExchangeCode());
		parm.setExchangeId(exchange.getId());
		parm.setRecordType(CollectRecord.RecordTypeConstants.NEWS);
		return parm;
	}

	private CollectRecord backRecord(Exchange exchange) {
		CollectRecord record = new CollectRecord();
		record.setCollectJob(collectJob);
		record.setExchangeCode(exchange.getExchangeCode());
		record.setExchangeId(exchange.getId());
		record.setRecordType(CollectRecord.RecordTypeConstants.NEWS);
		try {
			record.setCustomer(InetAddress.getLocalHost().getHostAddress());
		} catch (Exception e) {
		}
		return record;
	}

	public static class BloomTag {
		private Integer wbRegionId;
		private String bbRegionAlias;

		public Integer getWbRegionId() {
			return wbRegionId;
		}

		public void setWbRegionId(Integer wbRegionId) {
			this.wbRegionId = wbRegionId;
		}

		public String getBbRegionAlias() {
			return bbRegionAlias;
		}

		public void setBbRegionAlias(String bbRegionAlias) {
			this.bbRegionAlias = bbRegionAlias;
		}

		private static BloomTag produce(Integer wbRegionId, String BbRegionAlias) {
			BloomTag t = new BloomTag();
			t.wbRegionId = wbRegionId;
			t.bbRegionAlias = BbRegionAlias;
			return t;
		}
	}
}
